package com.ververica.cdc.guass.source.kafka.data;

import org.apache.flink.api.connector.source.ReaderOutput;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.table.data.RowData;

public  class ReaderOutputWrapper implements SourceFunction.SourceContext<RowData> {

    private final ReaderOutput<RowData> output;
    private final String splitId;

    public ReaderOutputWrapper(ReaderOutput<RowData> output, String splitId) {
        this.output = output;
        this.splitId = splitId;
    }

    @Override
    public void collect(RowData element) {
        output.collect(element);
    }

    @Override
    public void collectWithTimestamp(RowData element, long timestamp) {
        output.collect(element);
    }

    @Override
    public void emitWatermark(Watermark mark) {
        output.emitWatermark(new org.apache.flink.api.common.eventtime.Watermark(mark.getTimestamp()));
    }

    @Override
    public void markAsTemporarilyIdle() {
        output.markIdle();
    }

    @Override
    public Object getCheckpointLock() {
        return splitId;
    }

    @Override
    public void close() {
        // 关闭资源
        output.releaseOutputForSplit(splitId);
    }
}